package com.rocky.component.elasticsearch.service;

import com.rocky.component.elasticsearch.annotation.RockyESModel;
import com.rocky.component.elasticsearch.config.property.RockyElasticsearchProperty;
import com.rocky.component.elasticsearch.model.RockyDocumentWrapper;
import com.rocky.component.elasticsearch.service.event.RockyDocumentCallback;
import com.rocky.component.util.ObjectMapperUtil;
import lombok.RequiredArgsConstructor;
import org.apache.commons.lang3.StringUtils;
import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.action.DocWriteRequest;
import org.elasticsearch.action.DocWriteResponse;
import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.support.WriteRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.VersionType;
import org.elasticsearch.rest.RestStatus;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.HashMap;
import java.util.List;
import java.util.Objects;

/**
 * @author rocky
 * date 2025-08-09
 * Description:
 */
@RequiredArgsConstructor
public class RockyDocumentService {

    private final Logger logger = LoggerFactory.getLogger(RockyDocumentService.class);

    private final RestHighLevelClient restHighLevelClient;

    private final RockyElasticsearchProperty elasticsearchProperty;

    public boolean index(RockyDocumentWrapper wrapper) {
        String index = parseRockyESIndex(wrapper.getBody());

        return index(index, wrapper.getId(), wrapper.getVersion(), wrapper.getBody());
    }

    private boolean index(String index, String id, Long version, Object object) {
        return index(index, id, version, object, new RockyDocumentCallback() {
        });
    }

    private boolean index(String index, String id, Long version, Object body, RockyDocumentCallback documentCallback) {
        if (StringUtils.isBlank(index)) {
            throw new RuntimeException("Rocky elasticsearch index field cannot be null or empty");
        }
        if (StringUtils.isBlank(id)) {
            throw new RuntimeException("Rocky elasticsearch index [" + index + "], the document id field cannot be null or empty");
        }
        if (Objects.isNull(version)) {
            throw new RuntimeException("Rocky elasticsearch index [" + index + "], the document version field cannot be null");
        }
        if (Objects.isNull(body)) {
            throw new RuntimeException("Rocky elasticsearch index [" + index + "], the document body field cannot be null");
        }

        IndexRequest indexRequest = new IndexRequest(index)
                .id(id)
                .version(version)
                .source(ObjectMapperUtil.jsonString(body), XContentType.JSON)
                .versionType(VersionType.EXTERNAL_GTE)
                .setRefreshPolicy(WriteRequest.RefreshPolicy.WAIT_UNTIL);

        IndexResponse indexResponse;
        try {
            indexResponse = restHighLevelClient.index(indexRequest, RequestOptions.DEFAULT);
        } catch (ElasticsearchException e) {
            if (RestStatus.CONFLICT == e.status()) {
                logger.error("Rocky elasticsearch index [" + index + "], an error occurred while updating the document with ID [" + id + "] due to a version conflict", e);
                return false;
            }

            throw new RuntimeException(e);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }

        if (indexResponse.getResult() == DocWriteResponse.Result.CREATED) {
            documentCallback.onCreatedSuccess();
        } else if (indexResponse.getResult() == DocWriteResponse.Result.UPDATED) {
            documentCallback.onUpdatedSuccess();
        } else {
            return false;
        }

        return true;
    }

    public boolean delete(RockyDocumentWrapper wrapper) {
        String index = parseRockyESIndex(wrapper.getBody());

        return delete(index, wrapper.getId(), wrapper.getVersion());
    }

    private boolean delete(String index, String id, Long version) {
        if (StringUtils.isBlank(index)) {
            throw new RuntimeException("Rocky elasticsearch index field cannot be null or empty");
        }
        if (StringUtils.isBlank(id)) {
            throw new RuntimeException("Rocky elasticsearch index [" + index + "], the document id field cannot be null or empty");
        }
        if (Objects.isNull(version)) {
            version = System.currentTimeMillis();
        }

        DeleteRequest deleteRequest = new DeleteRequest(index, id)
                .version(version)
                .versionType(VersionType.EXTERNAL_GTE)
                .setRefreshPolicy(WriteRequest.RefreshPolicy.WAIT_UNTIL);

        DeleteResponse deleteResponse;
        try {
            deleteResponse = restHighLevelClient.delete(deleteRequest, RequestOptions.DEFAULT);
        } catch (ElasticsearchException e) {
            if (RestStatus.CONFLICT == e.status()) {
                logger.error("Rocky elasticsearch index [" + index + "], an error occurred while deleting the document with ID [" + id + "] due to a version conflict", e);
                return false;
            }

            throw new RuntimeException(e);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }

        if (deleteResponse.getResult() == DocWriteResponse.Result.NOT_FOUND) {
            logger.warn("Rocky elasticsearch index [" + index + "], the document [" + id + "] have been deleted or not found");
            return false;
        }

        return deleteResponse.getResult() == DocWriteResponse.Result.DELETED;
    }

    public boolean bulk(List<DocWriteRequest<?>> docWriteRequestList, RockyDocumentCallback rockyDocumentCallback) {
        BulkRequest bulkRequest = new BulkRequest()
                .add(docWriteRequestList)
                .setRefreshPolicy(WriteRequest.RefreshPolicy.WAIT_UNTIL);

        BulkResponse bulkResponse;
        try {
            bulkResponse = restHighLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        if (!bulkResponse.hasFailures()) {
            return true;
        }

        for (BulkItemResponse bulkItemResponse : bulkResponse) {
            if (bulkItemResponse.isFailed()) {
                rockyDocumentCallback.onBulkFailure(bulkItemResponse);
            }
        }

        return true;
    }

    private String parseRockyESIndex(Object object) {
        Class<?> aClass = object.getClass();
        RockyESModel rockyESModel = aClass.getAnnotation(RockyESModel.class);
        if (Objects.isNull(rockyESModel)) {
            throw new RuntimeException("RockyESModel annotation was not found on the class: " + aClass.getName());
        }

        String index = elasticsearchProperty.getIndex().getIndexPrefixName() + rockyESModel.indexName();
        if (StringUtils.isNotBlank(rockyESModel.indexNamePrefix())) {
            index = rockyESModel.indexNamePrefix() + rockyESModel.indexName();
        }

        return index;
    }
}
